package com.shujia.wyh.mr;


/*
    hello world java
    hello nihao hadoop
    hello java hive
    world hive


    hello 3
    world 2
    java 2
    nihao 1
    hadoop 1
    hive 2
 */

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapTask;
import org.apache.hadoop.mapred.ReduceTask;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCount {
    //创建Map类
    //KEYIN, VALUEIN, KEYOUT, VALUEOUT
    //hadoop环境中有着自己的一套数据类型
    //hadoop中规定，map输入的key类型是LongWritable,value值是Text
    //hello world java
    //<0,"hello world java">
    //<"hello",1>
    //<"world",1>
    //<"java",1>
    static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
        //map逻辑是作用在每一行上的
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
            //将Hadoop中的Text类型数据变成java中的String类型
            String s = value.toString();
            //将进来的一行数据，按照指定的分隔符进行分割
            String[] s1 = s.split(" ");  //["hello","world","java"]

            //context是hadoop提供给MR的上下文，可以通过这个context将数据写出到环形缓冲区中
            //遍历数组，得到每个单词
            for (String word : s1) {
                //将自己的String类型封装成目标Text类型
                context.write(new Text(word), new LongWritable(1L));
            }

//            context.write(new Text("【行值】" + key + "--【内容】" + value), new LongWritable(1L));
        }
    }

    //中间其实有进入环形缓冲区，分区编号，快速排序，归并排序，归并排序，这些其实都在底层给你执行完了
    //接下来只需要写reduce合并之后执行的逻辑就行了

//    static class MyCombiner extends Reducer<>{
//        @Override
//        protected void reduce(Object key, Iterable values, Reducer.Context context) throws IOException, InterruptedException {
//            //这是自己编写预聚合的逻辑
//        }
//    }

    //创建Reduce类
    static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
        //<"hello",[1,1,1,1,1,..]>
        //<>
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {
            //定义一个额外的变量接收总和
            long sum = 0;

            //相同的key分为一组，将value值相加
            //在java中，只有Collection集合才会有迭代器
            //又想到增强for循环是可以遍历迭代器的
            for (LongWritable value : values) {
                //因为此刻是LongWritable类型不能直接与long数据相加
                long l = value.get();
                sum += l;
            }

            //<"hello",3>
            //使用hadoop上下文对象将结果写入到磁盘
            context.write(key,new LongWritable(sum));

//            context.write(key, new LongWritable(1L));

        }
    }

    // hadoop jar xxx.jar xxx /xxx/xxx/xx.txt  /xxx/xxx/xxx
    public static void main(String[] args) throws Exception {
        //获取hadoop基本环境配置
        Configuration conf = new Configuration();
//        conf.set("","");

        //创建MapReduce作业
        Job job = Job.getInstance(conf);

        //给作业起一个名字，将来运行的时候在Yarn中能够看到
        job.setJobName("25期 WordCount案例");

        //设置reduce的个数
        //如果不设置，默认只有一个Reduce任务
//        job.setNumReduceTasks(2);

        //指定主类名
        job.setJarByClass(WordCount.class);

        //设置Map类，将来会去执行Map类中的map方法产生map任务
        job.setMapperClass(MyMapper.class);
        //设置Map逻辑出来的key，value的数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        //设置Reduce类，将来会去执行Reduce类中的reduce方法产生reduce任务
        job.setReducerClass(MyReducer.class);
        //设置Reduce逻辑出来的key，value的数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        //指定预聚合的类，一般情况下，是与Reduce同一个类
        job.setCombinerClass(MyReducer.class);

        //指定数据输入路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        //指定输出路径（输出的类型是文件夹，会自动创建，不需要手动创建）
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //提交作业到集群，开始执行
        job.waitForCompletion(true);

//        ReduceTask
    }
}
